package com.ironman.event.mqtt;


import com.ironman.common.threadpool.ThreadPool;
import com.ironman.event.properties.MqttRemoteProperties;
import com.ironman.event.properties.SysTrainingProperties;
import com.ironman.event.topic.dami.WindTopic;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.codec.MqttQoS;
import org.dromara.mica.mqtt.codec.MqttTopicSubscription;
import org.dromara.mica.mqtt.core.client.IMqttClientConnectListener;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.noear.dami.Dami;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.tio.core.ChannelContext;

/**
 * description mqtt 远程客户端
 * company 铁人科技
 *
 * @author 陈浩杰
 * @date 2024/3/24 10:10
 */
@Slf4j
@Configuration
public class RemoteMqttClientConfig {

    @Bean
    public MqttClient mqttClient(MqttRemoteProperties mqttRemoteProperties, SysTrainingProperties sysTrainingProperties) {
        return MqttClient.create()
                .name("sailboat-remote-mqtt-client")
                .ip(mqttRemoteProperties.getIp())
                .port(mqttRemoteProperties.getPort())
                .username(mqttRemoteProperties.getUsername())
                .password(mqttRemoteProperties.getPassword())
                .clientId(mqttRemoteProperties.getClientId())
                .keepAliveSecs(mqttRemoteProperties.getKeepAliveInterval())
                .retryCount(mqttRemoteProperties.getRetryCount())
                .globalSubscribe(new MqttTopicSubscription("gateway/env/wind/" + sysTrainingProperties.getTenantCode() + "/+", MqttQoS.valueOf(1)))
                .globalMessageListener((context, topic, message, payload) -> {
                    ThreadPool.FIXED_EXECUTOR.submit(() -> {
                        String msg = new String(payload);
                        Dami.<String, Void>bus().send(WindTopic.GW_WIND_TOPIC, msg);
                    });
                })
                .connectListener(new IMqttClientConnectListener() {
                    @Override
                    public void onConnected(ChannelContext context, boolean isReconnect) {
                        if (isReconnect) {
                            log.warn("重连 mqtt 服务器重连成功...");
                        } else {
                            log.info("连接 mqtt 服务器成功...");
                        }
                    }

                    @Override
                    public void onDisconnect(ChannelContext channelContext, Throwable throwable, String remark, boolean isRemove) {
                        log.error("mqtt 链接断开 remark:{} isRemove:{}", remark, isRemove, throwable);
                    }
                })
                .connect();
    }
}
